package com.leo.netty.simple;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;

import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * @author du
 */
public class NettyServer {

    public static void main(String[] args) {

        /**
         * 1. 创建BossGroup 和 WorkerGroup
         * boss group 只处理连接请求, worker group 只处理读写
         * 两个都是无限循环
         * 可以指定线程数量, 若不指定则默认为: cpu核数 * 2
         */

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            //2. 配置服务端启动对象
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup);
            bootstrap.option(ChannelOption.SO_BACKLOG, 128); // 设置线程队列得到连接个数
            bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true); //设置保持活动连接状态

            // boss group(服务端)使用的通道类型(反射), 可以是: ServerSocketChannel(netty重写过), NioServerSocketChannel
            bootstrap.channel(NioServerSocketChannel.class);
            bootstrap.attr(AttributeKey.newInstance("attr"), "attrValue");
            bootstrap.childAttr(AttributeKey.newInstance("childAttr"), "childAttrValue");
            //bootstrap.handler(new ServerHandler()); //设置服务端boss group的处理器
            // worker group使用的通道类型, 可以是: SocketChannel(netty重写过), NioSocketChannel
            bootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel socketChannel) throws Exception {
                    // channel和pipeline相互持有, 往pipeline添加处理器 (处理器链表)
                    /*socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(65535, Delimiters.lineDelimiter()[0]));
                    socketChannel.pipeline().addLast(new StringDecoder());
                    socketChannel.pipeline().addLast(new StringEncoder());*/
                    socketChannel.pipeline().addLast(new ServerHandler());
                }
            });


            //绑定端口并且同步, 启动服务
            ChannelFuture future = bootstrap.bind(8888).sync();
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        System.out.println("监听端口8888成功");
                    } else {
                        System.out.println("监听端口8888失败");
                    }
                }
            });
            System.out.println("服务端已启动...");
            //对关闭通道事件进行监听
            future.channel().closeFuture().sync();

        } catch (Exception e) {

        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    /**
     * 执行于连接初始化时
     * 1. 自定义handler需要继承Netty规定好的某个HandlerAdapter适配器
     * 2. ChannelInboundHandlerAdapter指的是: 请求入站处理器适配器规范
     */
    private static class ServerHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            //通道就绪
            System.out.println("channelActive");
        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) {
            System.out.println("channelRegistered");
        }

        @Override
        public void handlerAdded(ChannelHandlerContext ctx) {
            System.out.println("handlerAdded");
        }

        /**
         * 通道有读取事件时会触发
         * @param ctx ctx是一个上下文对象，包含了 pipeline/channel 对象的引用
         * @param msg 一般是ByteBuf对象
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("服务器 worker group 读取线程: " + Thread.currentThread().getName());
            Channel channel = ctx.channel();
            ChannelPipeline pipeline = ctx.pipeline();
            ChannelPipeline pipeline1 = channel.pipeline();
            Channel channel1 = pipeline.channel();
            System.out.println(channel == channel1); //true
            System.out.println(pipeline == pipeline1); //true

            ctx.channel().eventLoop().execute(() -> {
                try {
                    Thread.sleep(10 * 1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("用户自定义耗时10s任务1", CharsetUtil.UTF_8));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

            ctx.channel().eventLoop().execute(() -> {
                try {
                    Thread.sleep(5 * 1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("用户自定义耗时5s任务2", CharsetUtil.UTF_8));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

            ctx.channel().eventLoop().schedule(() -> {
                try {
                    Thread.sleep(10 * 1000);
                } catch (InterruptedException e) { }
                ctx.writeAndFlush(Unpooled.copiedBuffer("用户自定义耗时10s任务1", CharsetUtil.UTF_8));
            }, 5, TimeUnit.SECONDS);


            // 接收到的数据是一个ByteBuf对象, 需要强转
            ByteBuf buf = (ByteBuf) msg;

            System.out.println("客户端发送的消息是: " + buf.toString(CharsetUtil.UTF_8));
            System.out.println("客户端地址是: " + ctx.channel().remoteAddress());
        }

        /**
         * 数据读取完毕
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            //super.channelReadComplete(ctx);
            //给客户端回一个消息
            ctx.writeAndFlush(Unpooled.copiedBuffer("hello, client", CharsetUtil.UTF_8));
        }

        /**
         * 异常处理, 一般是关闭通道
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            //super.exceptionCaught(ctx, cause);
            ctx.channel().close();
            //ctx.close();
        }
    }
}
